DEMO CSV SQL Agent

CODE + DEMO
- GitHub code with video demo
Purpose
- Let users query CSV file in a natural way
  - Ex: what is the most expensive product and where is it from?
Workflow
- User first uploads the CSV file to process
  - The agent initiates the process file flow
    - CSV file is read and loaded to panda frames
    - We get CSV schema from pandas
      - We need this later
    - We then store the data to local sqlite db using pandas toSQL option
- User asks analytics question to Agent
  - Agent initiates find result flow:
    - Users natural question is turned to SQL
      - Schema is passed to the Agent to generate proper sql
    - Sub agent uses sql to result tool to execute the generated sql
    - DB Result is then refined in a language the user understands

Code Snippets

Main Agent(root agent)
- Root agents responsibility is to define our agent and acts as a delegator

root_agent = Agent(
    name="file_analytics_agent",
    model=AGENT_MODEL,
    instruction="""
You are CSV Analytics Agent (CAA), designed to process CSV files and perform data analytics.

Purpose:
- Process uploaded CSV files and store them in a local SQLite database
- Answer analytical questions about the stored data

Capabilities:
- File Processing: Handles CSV file uploads and stores data in SQLite
- Analytics: Converts natural language questions to SQL queries and returns formatted results
- Error Handling: Provides clear error messages for invalid inputs or operations

Examples:
- "Process this CSV file" → Processes and stores the file
- "What are the top-selling products?" → Analyzes data and returns results
- "Store this data" → Processes and stores the file
- If a person greets you asks for your introduction -> respond with who you are and your purpose

Limitations:
- Only handles CSV file processing and related analytics
- Cannot perform unrelated tasks (e.g., general knowledge questions)

""",
    description="Main agent that delegates tasks to appropriate sub-agents",
    tools=[
        file_processor_tool,
        AgentTool(agent=query_handler_agent)
    ],
)

File processor tool
- When a user uploads a CSV file this tool needs to process it and store the data to local sqlite db

def file_processor_tool(tool_context: ToolContext) -> Dict[str, str]:
    """
    Processes CSV files and stores them in a SQLite database.

    Returns:
        Dict[str, str]: A dictionary containing:
            - status: 'success' or 'error'
            - response: Descriptive message about the processing result
    """
    contents = tool_context.user_content
    df = None

    for part in contents.parts:
        if part.inline_data and part.inline_data.mime_type == "text/csv":
            try:
                binary_csv_data = part.inline_data.data
                decoded_data = binary_csv_data.decode('utf-8')
                string_buffer = io.StringIO(decoded_data)
                df = pd.read_csv(string_buffer)
            except Exception as e:
                return {
                    "status": "error",
                    "response": f"Failed to read CSV file: {str(e)}"
                }
    if df is None:
        return {
            "status": "error",
            "response": "Please provide a valid CSV file"
        }

    tool_context.state[TABLE_INFO_SESSION_KEY] =  f"Table name={TABLE_NAME}, columns info: {df.dtypes.to_string()}"

    try:
        conn = sqlite3.connect(DB_PATH)
        df.to_sql(
            name=TABLE_NAME,
            con=conn,
            if_exists='replace',
            index=False,
        )
        conn.close()
        return {
            "status": "success",
            "response": "File processed successfully. You can now ask analytical questions."
        }
    except Exception as e:
        return {
            "status": "error",
            "response": f"Failed to save data to SQLite: {str(e)}"
        }

Query handler agent
- To get user desired result we first need to convert the query to sql then execute the sql on our db

query_handler_agent = SequentialAgent(
    name="query_handler_agent",
    description="Handles user queries by converting them to SQL and formatting results",
    sub_agents=[
        question_to_sql_agent,
        sql_to_message_agent
    ]
)

Question to sql agent
- When generating sql we need to give the model some extra information such as the table schema, sample row values,…

question_to_sql_agent = Agent(
    model=AGENT_MODEL,
    name="question_to_sql_agent",
    instruction=f"""
Convert natural language requests into SQLite-compatible SQL queries.

Input:
- Natural language query from the user
- Table information stored in state under key: {{{TABLE_INFO_SESSION_KEY}}}

Requirements:
- Generate precise, executable SQLite queries
- Use table and column information from the state key {{{TABLE_INFO_SESSION_KEY}}}
- Return only the final SQL query string
- If conversion fails, return a clear error message describing the issue
- Ensure queries are safe and optimized for SQLite

Output:
- A single SQLite-compatible SQL query string
- Or an error message if the query cannot be generated
""",
    description="Converts user questions into SQLite-compatible SQL queries",
    output_key="sql_query",
)

SQL to message agent
- Our final message should be in a format that our non technical user understands

def run_query(sql: str) -> Dict[str, Any]:
    """
    Executes a SQL query on the local SQLite database.

    Args:
        sql (str): The SQL query to execute.

    Returns:
        Dict[str, Any]: A dictionary containing:
            - status: 'success' or 'error'
            - result: Query results or error message
    """
    try:
        conn = sqlite3.connect("local.db")
        cursor = conn.cursor()
        cursor.execute(sql)
        results = cursor.fetchall()
        conn.close()

        return {
            "status": "success",
            "result": results if results else "No results found"
        }
    except sqlite3.Error as e:
        return {
            "status": "error",
            "result": f"Query execution failed: {str(e)}"
        }


sql_to_message_agent = Agent(
    model=AGENT_MODEL,
    name="sql_to_message_agent",
    instruction="""
Convert SQL query results into human-readable responses.

Requirements:
- Use the `run_query` tool to execute the SQL query from the state under key `sql_query`
- Format results in a clear, user-friendly manner
- Use tables, lists, or descriptive text as appropriate
- Handle empty results gracefully
- Include relevant context from the query in the response

Output:
- A human-readable string containing the formatted results
""",
    description="Converts SQL query results into user-friendly messages",
    tools=[run_query]
)

Code Snippets​

Code Snippets